/*-------------------------------------------------------------------------
 *
 * hashovfl.c
 *	  Postgres哈希访问方法的溢出页面管理代码
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/hash/hashovfl.c
 *
 * NOTES
 *	  溢出页面看起来像普通的关系页面。
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/hash.h"
#include "access/hash_xlog.h"
#include "access/xloginsert.h"
#include "miscadmin.h"
#include "utils/rel.h"


static uint32 fc__hash_firstfreebit(uint32 fc_map);


/*
 * 将溢出页位号（其在自由页位图中的索引）转换为索引内的块号。
 */
static BlockNumber fc_bitno_to_blkno(HashMetaPage fc_metap, uint32 fc_ovflbitnum)
{
	uint32		fc_splitnum = fc_metap->hashm_ovflpoint;
	uint32		fc_i;

	/* 将零基位号转换为基于1的页号 */
	fc_ovflbitnum += 1;

	/* 确定此页的分割编号（必须 >= 1） */
	for (fc_i = 1;
		 fc_i < fc_splitnum && fc_ovflbitnum > fc_metap->hashm_spares[fc_i];
		 fc_i++)
		 /* 循环 */ ;

	/*
	 * 通过添加在此分割点之前存在的桶页数，将其转换为绝对页号。
	 */
	return (BlockNumber) (_hash_get_totalbuckets(fc_i) + fc_ovflbitnum);
}

/*
 * _hash_ovflblkno_to_bitno
 *
 * 将溢出页块号转换为自由页位图中的位号。
 */
uint32 _hash_ovflblkno_to_bitno(HashMetaPage fc_metap, BlockNumber fc_ovflblkno)
{
	uint32		fc_splitnum = fc_metap->hashm_ovflpoint;
	uint32		fc_i;
	uint32		fc_bitnum;

	/* 确定包含此页的分割编号 */
	for (fc_i = 1; fc_i <= fc_splitnum; fc_i++)
	{
		if (fc_ovflblkno <= (BlockNumber) _hash_get_totalbuckets(fc_i))
			break;				/* 哦，糟糕 */
		fc_bitnum = fc_ovflblkno - _hash_get_totalbuckets(fc_i);

		/*
		 * bitnum 必须大于在先前分割点添加的溢出页的数量。
		 * 此分割编号（i）处的溢出页（如果有）应从
		 * (_hash_get_totalbuckets(i) +
		 * metap->hashm_spares[i - 1] + 1) 开始。
		 */
		if (fc_bitnum > fc_metap->hashm_spares[fc_i - 1] &&
			fc_bitnum <= fc_metap->hashm_spares[fc_i])
			return fc_bitnum - 1;	/* -1 以将基于1转换为基于0 */
	}

	ereport(ERROR,
			(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
			 errmsg("invalid overflow block number %u", fc_ovflblkno)));
	return 0;					/* 保持编译器安静 */
}

/*
 *	_hash_addovflpage
 *
 * 向最后一个页面由 'buf' 指向的桶中添加一个溢出页。
 *
 * 进入时，调用者必须持有 'buf' 的固定，但没有锁定。 在退出之前
 * 将释放固定（我们假设调用者对 'buf' 不再感兴趣），
 * 如果没有要求保留。 仅对主桶保持固定。
 * 返回的溢出页将被锁定并处于写入锁定状态；保证是空的。
 *
 * 调用者必须在元页缓冲区上持有固定，但没有锁定。
 * 该缓冲区将保持在相同状态。
 *
 * 注意：由于多个进程可能并发执行此操作，
 * 不应假设返回的溢出页将是原始传递的 'buf' 的
 * 直接后继。 可能在其中添加了额外的溢出页到桶链中。
 */
Buffer _hash_addovflpage(Relation fc_rel, Buffer fc_metabuf, Buffer fc_buf, bool fc_retain_pin)
{
	Buffer		fc_ovflbuf;
	Page		fc_page;
	Page		fc_ovflpage;
	HashPageOpaque fc_pageopaque;
	HashPageOpaque fc_ovflopaque;
	HashMetaPage fc_metap;
	Buffer		fc_mapbuf = InvalidBuffer;
	Buffer		fc_newmapbuf = InvalidBuffer;
	BlockNumber fc_blkno;
	uint32		fc_orig_firstfree;
	uint32		fc_splitnum;
	uint32	   *fc_freep = NULL;
	uint32		fc_max_ovflpg;
	uint32		fc_bit;
	uint32		fc_bitmap_page_bit;
	uint32		fc_first_page;
	uint32		fc_last_bit;
	uint32		fc_last_page;
	uint32		fc_i,
				fc_j;
	bool		fc_page_found = false;

	/*
	 * 写锁定尾页。 在这里，我们需要维护锁定顺序，
	 * 首先获取桶尾页的锁，然后在元页上获取锁
	 * 以查找并锁定位图页，如果找到，则释放元页上的锁，
	 * 最后获取新溢出缓冲区的锁。
	 * 我们需要这种锁定顺序以避免与正在进行插入的后端
	 * 发生死锁。
	 *
	 * 注意：如果我们为获取溢出页制作两个WAL记录
	 * （一个用于分配溢出页，另一个用于将其添加到溢出桶链），
	 * 我们可以避免在这里锁定许多缓冲区。 然而， 
	 * 如果系统在分配后崩溃，这可能会泄漏一个溢出页。
	 * 不用说，从性能的角度来看，最好只有一个记录。
	 */
	LockBuffer(fc_buf, BUFFER_LOCK_EXCLUSIVE);

	/* 可能是冗余的... */
	_hash_checkpage(fc_rel, fc_buf, LH_BUCKET_PAGE | LH_OVERFLOW_PAGE);

	/* 循环以查找当前尾页，以防其他人也插入 */
	for (;;)
	{
		BlockNumber fc_nextblkno;

		fc_page = BufferGetPage(fc_buf);
		fc_pageopaque = HashPageGetOpaque(fc_page);
		fc_nextblkno = fc_pageopaque->hasho_nextblkno;

		if (!BlockNumberIsValid(fc_nextblkno))
			break;

		/* 我们假设我们不需要写入未修改的页面 */
		if (fc_retain_pin)
		{
			/* 固定将仅保留针对主桶页面 */
			Assert((fc_pageopaque->hasho_flag & LH_PAGE_TYPE) == LH_BUCKET_PAGE);
			LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);
		}
		else
			_hash_relbuf(fc_rel, fc_buf);

		fc_retain_pin = false;

		fc_buf = _hash_getbuf(fc_rel, fc_nextblkno, HASH_WRITE, LH_OVERFLOW_PAGE);
	}

	/* 获取元页的独占锁 */
	LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);

	_hash_checkpage(fc_rel, fc_metabuf, LH_META_PAGE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	/* 从 hashm_firstfree 开始搜索 */
	fc_orig_firstfree = fc_metap->hashm_firstfree;
	fc_first_page = fc_orig_firstfree >> BMPG_SHIFT(fc_metap);
	fc_bit = fc_orig_firstfree & BMPG_MASK(fc_metap);
	fc_i = fc_first_page;
	fc_j = fc_bit / BITS_PER_MAP;
	fc_bit &= ~(BITS_PER_MAP - 1);

	/* 外层循环每个位图页迭代一次 */
	for (;;)
	{
		BlockNumber fc_mapblkno;
		Page		fc_mappage;
		uint32		fc_last_inpage;

		/* 想要以最后一个现有的溢出页面结束搜索 */
		fc_splitnum = fc_metap->hashm_ovflpoint;
		fc_max_ovflpg = fc_metap->hashm_spares[fc_splitnum] - 1;
		fc_last_page = fc_max_ovflpg >> BMPG_SHIFT(fc_metap);
		fc_last_bit = fc_max_ovflpg & BMPG_MASK(fc_metap);

		if (fc_i > fc_last_page)
			break;

		Assert(fc_i < fc_metap->hashm_nmaps);
		fc_mapblkno = fc_metap->hashm_mapp[fc_i];

		if (fc_i == fc_last_page)
			fc_last_inpage = fc_last_bit;
		else
			fc_last_inpage = BMPGSZ_BIT(fc_metap) - 1;

		/* 在读取位图页面时释放对元页面的独占锁 */
		LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);

		fc_mapbuf = _hash_getbuf(fc_rel, fc_mapblkno, HASH_WRITE, LH_BITMAP_PAGE);
		fc_mappage = BufferGetPage(fc_mapbuf);
		fc_freep = HashPageGetBitmap(fc_mappage);

		for (; fc_bit <= fc_last_inpage; fc_j++, fc_bit += BITS_PER_MAP)
		{
			if (fc_freep[fc_j] != ALL_SET)
			{
				fc_page_found = true;

				/* 重新获取对元页面的独占锁 */
				LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);

				/* 将位转换为页面内的位编号 */
				fc_bit += fc__hash_firstfreebit(fc_freep[fc_j]);
				fc_bitmap_page_bit = fc_bit;

				/* 将位转换为绝对位编号 */
				fc_bit += (fc_i << BMPG_SHIFT(fc_metap));
				/* 计算回收溢出页面的地址 */
				fc_blkno = fc_bitno_to_blkno(fc_metap, fc_bit);

				/* 获取并初始化回收页面 */
				fc_ovflbuf = _hash_getinitbuf(fc_rel, fc_blkno);

				goto found;
			}
		}

		/* 此处没有空闲空间，尝试移动到下一个映射页面 */
		_hash_relbuf(fc_rel, fc_mapbuf);
		fc_mapbuf = InvalidBuffer;
		fc_i++;
		fc_j = 0;					/* 从下一个映射页面的起始处扫描 */
		fc_bit = 0;

		/* 重新获取对元页面的独占锁 */
		LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);
	}

	/*
	 * 没有空闲页面——必须扩展关系以添加溢出页面。
	 * 首先，检查我们是否也需要添加新的位图页面。
	 */
	if (fc_last_bit == (uint32) (BMPGSZ_BIT(fc_metap) - 1))
	{
		/*
		 * 我们创建新的位图页面，将所有页面标记为“正在使用”。
		 * 实际上，新位图范围内会立即存在两个页面：
		 * 位图页面本身，以及返回给调用者的下一个页面。
		 * 这两个页面都正确地标记为“正在使用”。
		 * 后续页面尚不存在，但预先将其标记为“正在使用”也是方便的。
		 */
		fc_bit = fc_metap->hashm_spares[fc_splitnum];

		/* 元页面已经有写锁 */
		if (fc_metap->hashm_nmaps >= HASH_MAX_BITMAPS)
			ereport(ERROR,
					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
					 errmsg("out of overflow pages in hash index \"%s\"",
							RelationGetRelationName(fc_rel))));

		fc_newmapbuf = _hash_getnewbuf(fc_rel, fc_bitno_to_blkno(fc_metap, fc_bit), MAIN_FORKNUM);
	}
	else
	{
		/*
		 * 这里没什么可做的；由于页面将在最后使用的页面之后，
		 * 我们知道它的位图位已预初始化为“正在使用”。
		 */
	}

	/* 计算新溢出页面的地址 */
	fc_bit = BufferIsValid(fc_newmapbuf) ?
		fc_metap->hashm_spares[fc_splitnum] + 1 : fc_metap->hashm_spares[fc_splitnum];
	fc_blkno = fc_bitno_to_blkno(fc_metap, fc_bit);

	/*
	 * 使用_hash_getnewbuf获取页面，以确保smgr对
	 * 关系长度的理解与我们的保持同步。 XXX 在持有元页面写锁的情况下
	 * 这样做令人讨厌；最好使用不会阻塞传入搜索的锁。
	 *
	 * 这里持有两个缓冲区锁（一个在桶的尾页上，
	 * 另一个在新溢出页面上）是可以的，因为没有其他人
	 * 在争抢对ovflbuf的访问。
	 */
	fc_ovflbuf = _hash_getnewbuf(fc_rel, fc_blkno, MAIN_FORKNUM);

found:

	/*
	 * 执行更新。在更改被记录之前不发出ereport(ERROR)。我们希望
	 * 一起记录位图页面和溢出页面的更改，以避免在添加新页面的情况下
	 * 丢失页面。
	 */
	START_CRIT_SECTION();

	if (fc_page_found)
	{
		Assert(BufferIsValid(fc_mapbuf));

		/* 在位图中将页面标记为“正在使用” */
		SETBIT(fc_freep, fc_bitmap_page_bit);
		MarkBufferDirty(fc_mapbuf);
	}
	else
	{
		/* 更新计数以指示新溢出页面已添加 */
		fc_metap->hashm_spares[fc_splitnum]++;

		if (BufferIsValid(fc_newmapbuf))
		{
			_hash_initbitmapbuffer(fc_newmapbuf, fc_metap->hashm_bmsize, false);
			MarkBufferDirty(fc_newmapbuf);

			/* 将新位图页面添加到元页面的位图列表中 */
			fc_metap->hashm_mapp[fc_metap->hashm_nmaps] = BufferGetBlockNumber(fc_newmapbuf);
			fc_metap->hashm_nmaps++;
			fc_metap->hashm_spares[fc_splitnum]++;
		}

		MarkBufferDirty(fc_metabuf);

		/*
		 * 对于新的溢出页面，我们不需要在位图页面中显式设置位，
		 * 因为默认情况下将其设置为“正在使用”。
		 */
	}

	/*
	 * 调整hashm_firstfree以避免冗余搜索。但如果有人在我们搜索位图页面时
	 * 移动了它，就不要冒险改变它。
	 */
	if (fc_metap->hashm_firstfree == fc_orig_firstfree)
	{
		fc_metap->hashm_firstfree = fc_bit + 1;
		MarkBufferDirty(fc_metabuf);
	}

	/* 初始化新的溢出页面 */
	fc_ovflpage = BufferGetPage(fc_ovflbuf);
	fc_ovflopaque = HashPageGetOpaque(fc_ovflpage);
	fc_ovflopaque->hasho_prevblkno = BufferGetBlockNumber(fc_buf);
	fc_ovflopaque->hasho_nextblkno = InvalidBlockNumber;
	fc_ovflopaque->hasho_bucket = fc_pageopaque->hasho_bucket;
	fc_ovflopaque->hasho_flag = LH_OVERFLOW_PAGE;
	fc_ovflopaque->hasho_page_id = HASHO_PAGE_ID;

	MarkBufferDirty(fc_ovflbuf);

	/* 逻辑上将溢出页面链接到前一个页面 */
	fc_pageopaque->hasho_nextblkno = BufferGetBlockNumber(fc_ovflbuf);

	MarkBufferDirty(fc_buf);

	/* XLOG 相关内容 */
	if (RelationNeedsWAL(fc_rel))
	{
		XLogRecPtr	fc_recptr;
		xl_hash_add_ovfl_page fc_xlrec;

		fc_xlrec.bmpage_found = fc_page_found;
		fc_xlrec.bmsize = fc_metap->hashm_bmsize;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, SizeOfHashAddOvflPage);

		XLogRegisterBuffer(0, fc_ovflbuf, REGBUF_WILL_INIT);
		XLogRegisterBufData(0, (char *) &fc_pageopaque->hasho_bucket, sizeof(Bucket));

		XLogRegisterBuffer(1, fc_buf, REGBUF_STANDARD);

		if (BufferIsValid(fc_mapbuf))
		{
			XLogRegisterBuffer(2, fc_mapbuf, REGBUF_STANDARD);
			XLogRegisterBufData(2, (char *) &fc_bitmap_page_bit, sizeof(uint32));
		}

		if (BufferIsValid(fc_newmapbuf))
			XLogRegisterBuffer(3, fc_newmapbuf, REGBUF_WILL_INIT);

		XLogRegisterBuffer(4, fc_metabuf, REGBUF_STANDARD);
		XLogRegisterBufData(4, (char *) &fc_metap->hashm_firstfree, sizeof(uint32));

		fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_ADD_OVFL_PAGE);

		PageSetLSN(BufferGetPage(fc_ovflbuf), fc_recptr);
		PageSetLSN(BufferGetPage(fc_buf), fc_recptr);

		if (BufferIsValid(fc_mapbuf))
			PageSetLSN(BufferGetPage(fc_mapbuf), fc_recptr);

		if (BufferIsValid(fc_newmapbuf))
			PageSetLSN(BufferGetPage(fc_newmapbuf), fc_recptr);

		PageSetLSN(BufferGetPage(fc_metabuf), fc_recptr);
	}

	END_CRIT_SECTION();

	if (fc_retain_pin)
		LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);
	else
		_hash_relbuf(fc_rel, fc_buf);

	if (BufferIsValid(fc_mapbuf))
		_hash_relbuf(fc_rel, fc_mapbuf);

	LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);

	if (BufferIsValid(fc_newmapbuf))
		_hash_relbuf(fc_rel, fc_newmapbuf);

	return fc_ovflbuf;
}

/*
 *	_hash_firstfreebit()
 *
 *	返回在字‘map’中未设置的第一个位的编号。
 */
static uint32 fc__hash_firstfreebit(uint32 fc_map)
{
	uint32		fc_i,
				fc_mask;

	fc_mask = 0x1;
	for (fc_i = 0; fc_i < BITS_PER_MAP; fc_i++)
	{
		if (!(fc_mask & fc_map))
			return fc_i;
		fc_mask <<= 1;
	}

	elog(ERROR, "firstfreebit found no free bit");

	return 0;					/* 保持编译器安静 */
}

/*
 *	_hash_freeovflpage() -
 *
 *	从其桶链中移除此溢出页面，并将页面标记为
 *	可用。在进入时，ovflbuf 被写锁定；在退出前释放它。
 *
 *	在此函数中将元组 (itups) 添加到 wbuf。我们也可以在
 *	调用者中做到这一点，但在这里执行的好处是我们可以轻松地为
 *	XLOG_HASH_SQUEEZE_PAGE 操作编写 WAL。元组的添加和
 *	溢出页面的移除必须作为原子操作进行，否则
 *	在备用机上重放时，用户可能会发现重复记录。
 *
 *	由于此函数在 VACUUM 中调用，因此我们提供一个访问策略
 *	参数，该参数控制桶页面的抓取。
 *
 *	返回桶中给定页面后面页面的块编号，
 *	如果没有后续页面则返回 InvalidBlockNumber。
 *
 *	注意：调用者不能在元页面上保持锁定，也不能在
 *	桶链中紧邻 ovflbuf 的页面上保持锁定。如果在链中
 *	它与 wbuf 相同，我们不获取在 ovflbuf 之前的页面上的锁，
 *	因为调用者已经在相同页面上有锁。
 */
BlockNumber _hash_freeovflpage(Relation fc_rel, Buffer fc_bucketbuf, Buffer fc_ovflbuf,
				   Buffer fc_wbuf, IndexTuple *fc_itups, OffsetNumber *fc_itup_offsets,
				   Size *fc_tups_size, uint16 fc_nitups,
				   BufferAccessStrategy fc_bstrategy)
{
	HashMetaPage fc_metap;
	Buffer		fc_metabuf;
	Buffer		fc_mapbuf;
	BlockNumber fc_ovflblkno;
	BlockNumber fc_prevblkno;
	BlockNumber fc_blkno;
	BlockNumber fc_nextblkno;
	BlockNumber fc_writeblkno;
	HashPageOpaque fc_ovflopaque;
	Page		fc_ovflpage;
	Page		fc_mappage;
	uint32	   *fc_freep;
	uint32		fc_ovflbitno;
	int32		fc_bitmappage,
				fc_bitmapbit;
	Bucket		fc_bucket PG_USED_FOR_ASSERTS_ONLY;
	Buffer		fc_prevbuf = InvalidBuffer;
	Buffer		fc_nextbuf = InvalidBuffer;
	bool		fc_update_metap = false;

	/* 从 doomed 页面获取信息 */
	_hash_checkpage(fc_rel, fc_ovflbuf, LH_OVERFLOW_PAGE);
	fc_ovflblkno = BufferGetBlockNumber(fc_ovflbuf);
	fc_ovflpage = BufferGetPage(fc_ovflbuf);
	fc_ovflopaque = HashPageGetOpaque(fc_ovflpage);
	fc_nextblkno = fc_ovflopaque->hasho_nextblkno;
	fc_prevblkno = fc_ovflopaque->hasho_prevblkno;
	fc_writeblkno = BufferGetBlockNumber(fc_wbuf);
	fc_bucket = fc_ovflopaque->hasho_bucket;

	/*
	 * 修复桶链。 这是一个双向链表，因此我们必须修复
	 * 在被删除的溢出页面前后的桶链成员。
	 * 通过使用锁链来避免并发问题，具体描述见 hashbucketcleanup。
	 */
	if (BlockNumberIsValid(fc_prevblkno))
	{
		if (fc_prevblkno == fc_writeblkno)
			fc_prevbuf = fc_wbuf;
		else
			fc_prevbuf = _hash_getbuf_with_strategy(fc_rel,
												 fc_prevblkno,
												 HASH_WRITE,
												 LH_BUCKET_PAGE | LH_OVERFLOW_PAGE,
												 fc_bstrategy);
	}
	if (BlockNumberIsValid(fc_nextblkno))
		fc_nextbuf = _hash_getbuf_with_strategy(fc_rel,
											 fc_nextblkno,
											 HASH_WRITE,
											 LH_OVERFLOW_PAGE,
											 fc_bstrategy);

	/* 注意：bstrategy 故意不用于元页面和位图 */

	/* 读取元页面，以便确定使用哪个位图页面 */
	fc_metabuf = _hash_getbuf(fc_rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	/* 确定要设置的位 */
	fc_ovflbitno = _hash_ovflblkno_to_bitno(fc_metap, fc_ovflblkno);

	fc_bitmappage = fc_ovflbitno >> BMPG_SHIFT(fc_metap);
	fc_bitmapbit = fc_ovflbitno & BMPG_MASK(fc_metap);

	if (fc_bitmappage >= fc_metap->hashm_nmaps)
		elog(ERROR, "invalid overflow bit number %u", fc_ovflbitno);
	fc_blkno = fc_metap->hashm_mapp[fc_bitmappage];

	/* 在访问位图页面时释放元页面锁 */
	LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);

	/* 读取位图页面以清除位图位 */
	fc_mapbuf = _hash_getbuf(fc_rel, fc_blkno, HASH_WRITE, LH_BITMAP_PAGE);
	fc_mappage = BufferGetPage(fc_mapbuf);
	fc_freep = HashPageGetBitmap(fc_mappage);
	Assert(ISSET(fc_freep, fc_bitmapbit));

	/* 获取元页面的写锁以更新 firstfree */
	LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);

	/* 此操作需要记录多个元组，为此准备 WAL */
	if (RelationNeedsWAL(fc_rel))
		XLogEnsureRecordSpace(HASH_XLOG_FREE_OVFL_BUFS, 4 + fc_nitups);

	START_CRIT_SECTION();

	/*
	 * 我们必须在 "写入" 页面上插入元组，注意保持
	 * hashkey 排序。（如果我们向同一 "写入" 页面插入许多元组，
	 * 则值得对它们进行 qsort）。
	 */
	if (fc_nitups > 0)
	{
		_hash_pgaddmultitup(fc_rel, fc_wbuf, fc_itups, fc_itup_offsets, fc_nitups);
		MarkBufferDirty(fc_wbuf);
	}

	/*
	 * 重新初始化已释放的溢出页面。仅将页面归零是无效的，
	 * 因为 WAL 重放例程期望页面被初始化。见
	 * XLogReadBufferExtended 上 RBM_NORMAL 模式的解释。我们
	 * 小心地使特殊空间在这里有效，以便像
	 * pageinspect 这样的工具不会感到困惑。
	 */
	_hash_pageinit(fc_ovflpage, BufferGetPageSize(fc_ovflbuf));

	fc_ovflopaque = HashPageGetOpaque(fc_ovflpage);

	fc_ovflopaque->hasho_prevblkno = InvalidBlockNumber;
	fc_ovflopaque->hasho_nextblkno = InvalidBlockNumber;
	fc_ovflopaque->hasho_bucket = InvalidBucket;
	fc_ovflopaque->hasho_flag = LH_UNUSED_PAGE;
	fc_ovflopaque->hasho_page_id = HASHO_PAGE_ID;

	MarkBufferDirty(fc_ovflbuf);

	if (BufferIsValid(fc_prevbuf))
	{
		Page		fc_prevpage = BufferGetPage(fc_prevbuf);
		HashPageOpaque fc_prevopaque = HashPageGetOpaque(fc_prevpage);

		Assert(fc_prevopaque->hasho_bucket == fc_bucket);
		fc_prevopaque->hasho_nextblkno = fc_nextblkno;
		MarkBufferDirty(fc_prevbuf);
	}
	if (BufferIsValid(fc_nextbuf))
	{
		Page		fc_nextpage = BufferGetPage(fc_nextbuf);
		HashPageOpaque fc_nextopaque = HashPageGetOpaque(fc_nextpage);

		Assert(fc_nextopaque->hasho_bucket == fc_bucket);
		fc_nextopaque->hasho_prevblkno = fc_prevblkno;
		MarkBufferDirty(fc_nextbuf);
	}

	/* 清除位图位以指示此溢出页面是可用的 */
	CLRBIT(fc_freep, fc_bitmapbit);
	MarkBufferDirty(fc_mapbuf);

	/* 如果这是现在的第一个空闲页面，更新 hashm_firstfree */
	if (fc_ovflbitno < fc_metap->hashm_firstfree)
	{
		fc_metap->hashm_firstfree = fc_ovflbitno;
		fc_update_metap = true;
		MarkBufferDirty(fc_metabuf);
	}

	/* XLOG 相关内容 */
	if (RelationNeedsWAL(fc_rel))
	{
		xl_hash_squeeze_page fc_xlrec;
		XLogRecPtr	fc_recptr;
		int			fc_i;

		fc_xlrec.prevblkno = fc_prevblkno;
		fc_xlrec.nextblkno = fc_nextblkno;
		fc_xlrec.ntups = fc_nitups;
		fc_xlrec.is_prim_bucket_same_wrt = (fc_wbuf == fc_bucketbuf);
		fc_xlrec.is_prev_bucket_same_wrt = (fc_wbuf == fc_prevbuf);

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, SizeOfHashSqueezePage);

		/*
		 * 桶缓冲区需要注册，以确保我们能够在重放期间在其上获取
		 * 清理锁。
		 */
		if (!fc_xlrec.is_prim_bucket_same_wrt)
			XLogRegisterBuffer(0, fc_bucketbuf, REGBUF_STANDARD | REGBUF_NO_IMAGE);

		XLogRegisterBuffer(1, fc_wbuf, REGBUF_STANDARD);
		if (fc_xlrec.ntups > 0)
		{
			XLogRegisterBufData(1, (char *) fc_itup_offsets,
								fc_nitups * sizeof(OffsetNumber));
			for (fc_i = 0; fc_i < fc_nitups; fc_i++)
				XLogRegisterBufData(1, (char *) fc_itups[fc_i], fc_tups_size[fc_i]);
		}

		XLogRegisterBuffer(2, fc_ovflbuf, REGBUF_STANDARD);

		/*
		 * 如果 prevpage 和写页面（我们将元组从溢出移动到的块）
		 * 相同，则无需单独注册 prevpage。 在重放期间，
		 * 我们可以直接更新 writepage 中的 nextblock。
		 */
		if (BufferIsValid(fc_prevbuf) && !fc_xlrec.is_prev_bucket_same_wrt)
			XLogRegisterBuffer(3, fc_prevbuf, REGBUF_STANDARD);

		if (BufferIsValid(fc_nextbuf))
			XLogRegisterBuffer(4, fc_nextbuf, REGBUF_STANDARD);

		XLogRegisterBuffer(5, fc_mapbuf, REGBUF_STANDARD);
		XLogRegisterBufData(5, (char *) &fc_bitmapbit, sizeof(uint32));

		if (fc_update_metap)
		{
			XLogRegisterBuffer(6, fc_metabuf, REGBUF_STANDARD);
			XLogRegisterBufData(6, (char *) &fc_metap->hashm_firstfree, sizeof(uint32));
		}

		fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_SQUEEZE_PAGE);

		PageSetLSN(BufferGetPage(fc_wbuf), fc_recptr);
		PageSetLSN(BufferGetPage(fc_ovflbuf), fc_recptr);

		if (BufferIsValid(fc_prevbuf) && !fc_xlrec.is_prev_bucket_same_wrt)
			PageSetLSN(BufferGetPage(fc_prevbuf), fc_recptr);
		if (BufferIsValid(fc_nextbuf))
			PageSetLSN(BufferGetPage(fc_nextbuf), fc_recptr);

		PageSetLSN(BufferGetPage(fc_mapbuf), fc_recptr);

		if (fc_update_metap)
			PageSetLSN(BufferGetPage(fc_metabuf), fc_recptr);
	}

	END_CRIT_SECTION();

	/* 如果先前桶与写入桶不同，则释放先前桶 */
	if (BufferIsValid(fc_prevbuf) && fc_prevblkno != fc_writeblkno)
		_hash_relbuf(fc_rel, fc_prevbuf);

	if (BufferIsValid(fc_ovflbuf))
		_hash_relbuf(fc_rel, fc_ovflbuf);

	if (BufferIsValid(fc_nextbuf))
		_hash_relbuf(fc_rel, fc_nextbuf);

	_hash_relbuf(fc_rel, fc_mapbuf);
	_hash_relbuf(fc_rel, fc_metabuf);

	return fc_nextblkno;
}


/*
 * _hash_initbitmapbuffer()
 *
 * 初始化一个新的位图页面。新位图页面中的所有位都被设置为“1”，表示“正在使用”。
 */
void _hash_initbitmapbuffer(Buffer fc_buf, uint16 fc_bmsize, bool fc_initpage)
{
	Page		fc_pg;
	HashPageOpaque fc_op;
	uint32	   *fc_freep;

	fc_pg = BufferGetPage(fc_buf);

	/* 初始化页面 */
	if (fc_initpage)
		_hash_pageinit(fc_pg, BufferGetPageSize(fc_buf));

	/* 初始化页面的特殊空间 */
	fc_op = HashPageGetOpaque(fc_pg);
	fc_op->hasho_prevblkno = InvalidBlockNumber;
	fc_op->hasho_nextblkno = InvalidBlockNumber;
	fc_op->hasho_bucket = InvalidBucket;
	fc_op->hasho_flag = LH_BITMAP_PAGE;
	fc_op->hasho_page_id = HASHO_PAGE_ID;

	/* 将所有位设置为1 */
	fc_freep = HashPageGetBitmap(fc_pg);
	MemSet(fc_freep, 0xFF, fc_bmsize);

	/*
 * 将pd_lower设置在位图页面数据的末尾之后。我们甚至可以将pd_lower设置为pd_upper，但这样更精确，并使页面看起来可以压缩为xlog.c。
 */
	((PageHeader) fc_pg)->pd_lower = ((char *) fc_freep + fc_bmsize) - (char *) fc_pg;
}


/*
 * _hash_squeezebucket(rel, bucket)
 *
 * 尝试将元组挤压到桶链中更早出现的页面，以试图释放溢出页面。当我们开始“挤压”时，开始取元组的页面（“读取”页面）是桶链中的最后一个桶，而我们开始挤压元组的页面（“写入”页面）是桶链中的第一页。读取页面向后工作，写入页面向前工作；当读取页面和写入页面是同一页面时，过程终止。
 *
 * 此过程完成时，保证桶中的所有页面都非空，除非桶完全为空（在这种情况下所有溢出页面将被释放）。原始实现要求在入口时也要满足这一要求，但对于调用者来说，让空的溢出页面留着，让这个函数来清理要容易得多。
 *
 * 调用者必须在目标桶的主页面上获取清理锁，以排除任何正在进行的扫描，这样可能会因为我们在这里进行的重排而容易混淆，导致返回相同的元组多次或某些元组根本不返回。
 * 为了防止任何并发扫描交叉挤压扫描，我们使用类似于hashbucketcleanup的锁链。请参阅hashbucketcleanup顶部的评论。
 *
 * 我们需要在主桶上保留一个引脚，以确保没有并发拆分可以开始。
 *
 * 由于此函数在VACUUM中被调用，我们提供一个访问策略参数，用于控制桶页面的提取。
 */
void _hash_squeezebucket(Relation fc_rel,
					Bucket fc_bucket,
					BlockNumber fc_bucket_blkno,
					Buffer fc_bucket_buf,
					BufferAccessStrategy fc_bstrategy)
{
	BlockNumber fc_wblkno;
	BlockNumber fc_rblkno;
	Buffer		fc_wbuf;
	Buffer		fc_rbuf;
	Page		fc_wpage;
	Page		fc_rpage;
	HashPageOpaque fc_wopaque;
	HashPageOpaque fc_ropaque;

	/*
 * 开始挤压到主桶页面。
 */
	fc_wblkno = fc_bucket_blkno;
	fc_wbuf = fc_bucket_buf;
	fc_wpage = BufferGetPage(fc_wbuf);
	fc_wopaque = HashPageGetOpaque(fc_wpage);

	/*
 * 如果没有溢出页面，就没有什么可以挤压。调用者负责释放主桶页面上的引脚。
 */
	if (!BlockNumberIsValid(fc_wopaque->hasho_nextblkno))
	{
		LockBuffer(fc_wbuf, BUFFER_LOCK_UNLOCK);
		return;
	}

	/*
 * 从基础桶页面开始，向前查找桶链中的最后一个页面。注意：我们假设哈希桶链通常小于VACUUM正在使用的缓冲环，否则在这里使用访问策略会适得其反。
 */
	fc_rbuf = InvalidBuffer;
	fc_ropaque = fc_wopaque;
	do
	{
		fc_rblkno = fc_ropaque->hasho_nextblkno;
		if (fc_rbuf != InvalidBuffer)
			_hash_relbuf(fc_rel, fc_rbuf);
		fc_rbuf = _hash_getbuf_with_strategy(fc_rel,
										  fc_rblkno,
										  HASH_WRITE,
										  LH_OVERFLOW_PAGE,
										  fc_bstrategy);
		fc_rpage = BufferGetPage(fc_rbuf);
		fc_ropaque = HashPageGetOpaque(fc_rpage);
		Assert(fc_ropaque->hasho_bucket == fc_bucket);
	} while (BlockNumberIsValid(fc_ropaque->hasho_nextblkno));

	/*
 * 挤压元组。
 */
	for (;;)
	{
		OffsetNumber fc_roffnum;
		OffsetNumber fc_maxroffnum;
		OffsetNumber fc_deletable[MaxOffsetNumber];
		IndexTuple	fc_itups[MaxIndexTuplesPerPage];
		Size		fc_tups_size[MaxIndexTuplesPerPage];
		OffsetNumber fc_itup_offsets[MaxIndexTuplesPerPage];
		uint16		fc_ndeletable = 0;
		uint16		fc_nitups = 0;
		Size		fc_all_tups_size = 0;
		int			fc_i;
		bool		fc_retain_pin = false;

readpage:
		/* 扫描“读取”页面中的每个元组 */
		fc_maxroffnum = PageGetMaxOffsetNumber(fc_rpage);
		for (fc_roffnum = FirstOffsetNumber;
			 fc_roffnum <= fc_maxroffnum;
			 fc_roffnum = OffsetNumberNext(fc_roffnum))
		{
			IndexTuple	fc_itup;
			Size		fc_itemsz;

			/* 跳过死元组 */
			if (ItemIdIsDead(PageGetItemId(fc_rpage, fc_roffnum)))
				continue;

			fc_itup = (IndexTuple) PageGetItem(fc_rpage,
											PageGetItemId(fc_rpage, fc_roffnum));
			fc_itemsz = IndexTupleSize(fc_itup);
			fc_itemsz = MAXALIGN(fc_itemsz);

			/*
 * 沿着桶链向上走，寻找足够大的页面来容纳这个项目和所有其他累积项目。如果到达读取页面则退出。
 */
			while (PageGetFreeSpaceForMultipleTuples(fc_wpage, fc_nitups + 1) < (fc_all_tups_size + fc_itemsz))
			{
				Buffer		fc_next_wbuf = InvalidBuffer;
				bool		fc_tups_moved = false;

				Assert(!PageIsEmpty(fc_wpage));

				if (fc_wblkno == fc_bucket_blkno)
					fc_retain_pin = true;

				fc_wblkno = fc_wopaque->hasho_nextblkno;
				Assert(BlockNumberIsValid(fc_wblkno));

				/* 如果我们到达读取页面，就不需要移动到下一页 */
				if (fc_wblkno != fc_rblkno)
					fc_next_wbuf = _hash_getbuf_with_strategy(fc_rel,
														   fc_wblkno,
														   HASH_WRITE,
														   LH_OVERFLOW_PAGE,
														   fc_bstrategy);

				if (fc_nitups > 0)
				{
					Assert(fc_nitups == fc_ndeletable);

					/*
					 * 该操作需要记录多个元组，为此准备
					 * WAL。
					 */
					if (RelationNeedsWAL(fc_rel))
						XLogEnsureRecordSpace(0, 3 + fc_nitups);

					START_CRIT_SECTION();

					/*
					 * 我们必须在“写”页面上插入元组，注意保持 hashkey 排序。 （如果我们将许多元组插入到同一个“写”页面中，值得对它们进行 qsort）。
					 */
					_hash_pgaddmultitup(fc_rel, fc_wbuf, fc_itups, fc_itup_offsets, fc_nitups);
					MarkBufferDirty(fc_wbuf);

					/* 删除我们已经移出读取页面的元组 */
					PageIndexMultiDelete(fc_rpage, fc_deletable, fc_ndeletable);
					MarkBufferDirty(fc_rbuf);

					/* XLOG 相关内容 */
					if (RelationNeedsWAL(fc_rel))
					{
						XLogRecPtr	fc_recptr;
						xl_hash_move_page_contents fc_xlrec;

						fc_xlrec.ntups = fc_nitups;
						fc_xlrec.is_prim_bucket_same_wrt = (fc_wbuf == fc_bucket_buf);

						XLogBeginInsert();
						XLogRegisterData((char *) &fc_xlrec, SizeOfHashMovePageContents);

						/*
						 * bucket 缓冲区需要注册，以确保我们在重放期间能够获取其清理锁。
						 */
						if (!fc_xlrec.is_prim_bucket_same_wrt)
							XLogRegisterBuffer(0, fc_bucket_buf, REGBUF_STANDARD | REGBUF_NO_IMAGE);

						XLogRegisterBuffer(1, fc_wbuf, REGBUF_STANDARD);
						XLogRegisterBufData(1, (char *) fc_itup_offsets,
											fc_nitups * sizeof(OffsetNumber));
						for (fc_i = 0; fc_i < fc_nitups; fc_i++)
							XLogRegisterBufData(1, (char *) fc_itups[fc_i], fc_tups_size[fc_i]);

						XLogRegisterBuffer(2, fc_rbuf, REGBUF_STANDARD);
						XLogRegisterBufData(2, (char *) fc_deletable,
											fc_ndeletable * sizeof(OffsetNumber));

						fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_MOVE_PAGE_CONTENTS);

						PageSetLSN(BufferGetPage(fc_wbuf), fc_recptr);
						PageSetLSN(BufferGetPage(fc_rbuf), fc_recptr);
					}

					END_CRIT_SECTION();

					fc_tups_moved = true;
				}

				/*
				 * 在获取下一个页面的锁后释放前一个页面的锁
				 */
				if (fc_retain_pin)
					LockBuffer(fc_wbuf, BUFFER_LOCK_UNLOCK);
				else
					_hash_relbuf(fc_rel, fc_wbuf);

				/* 如果我们到达读取页面，则没有更多操作要做 */
				if (fc_rblkno == fc_wblkno)
				{
					_hash_relbuf(fc_rel, fc_rbuf);
					return;
				}

				fc_wbuf = fc_next_wbuf;
				fc_wpage = BufferGetPage(fc_wbuf);
				fc_wopaque = HashPageGetOpaque(fc_wpage);
				Assert(fc_wopaque->hasho_bucket == fc_bucket);
				fc_retain_pin = false;

				/* 保持整洁 */
				for (fc_i = 0; fc_i < fc_nitups; fc_i++)
					pfree(fc_itups[fc_i]);
				fc_nitups = 0;
				fc_all_tups_size = 0;
				fc_ndeletable = 0;

				/*
				 * 在移动元组后，rpage 会被压缩，
				 * 因此我们需要对其重新扫描。
				 */
				if (fc_tups_moved)
					goto readpage;
			}

			/* 记住要从“读取”页面中删除的元组 */
			fc_deletable[fc_ndeletable++] = fc_roffnum;

			/*
			 * 我们需要一个索引元组的副本，因为它们可能作为溢出页面的一部分被释放，但我们需要它们来在 
			 * _hash_freeovflpage 中写入 WAL 记录。
			 */
			fc_itups[fc_nitups] = CopyIndexTuple(fc_itup);
			fc_tups_size[fc_nitups++] = fc_itemsz;
			fc_all_tups_size += fc_itemsz;
		}

		/*
		 * 如果我们到达这里，则“读取”页上没有活动元组 ---
		 * 当我们到达时它是空的，或者我们移动了所有元组。 因此我们可以
		 * 直接释放页面，而无需逐个删除元组。 然后推进到前一个“读取”页面。
		 *
		 * 这里的棘手点：如果我们的读取和写入页面在 bucket 链中是相邻的，
		 * 我们对 wbuf 的写锁将与
		 * _hash_freeovflpage 更新被删除页面的兄弟链接的尝试产生冲突。 在这种情况下，我们不需要再次加锁。
		 */
		fc_rblkno = fc_ropaque->hasho_prevblkno;
		Assert(BlockNumberIsValid(fc_rblkno));

		/* 释放此溢出页面（释放 rbuf） */
		_hash_freeovflpage(fc_rel, fc_bucket_buf, fc_rbuf, fc_wbuf, fc_itups, fc_itup_offsets,
						   fc_tups_size, fc_nitups, fc_bstrategy);

		/* 保持整洁 */
		for (fc_i = 0; fc_i < fc_nitups; fc_i++)
			pfree(fc_itups[fc_i]);

		/* 我们是在释放与 wbuf 相邻的页面吗？ */
		if (fc_rblkno == fc_wblkno)
		{
			/* 在桶扫描结束之前保持主桶页面的引脚 */
			if (fc_wblkno == fc_bucket_blkno)
				LockBuffer(fc_wbuf, BUFFER_LOCK_UNLOCK);
			else
				_hash_relbuf(fc_rel, fc_wbuf);
			return;
		}

		fc_rbuf = _hash_getbuf_with_strategy(fc_rel,
										  fc_rblkno,
										  HASH_WRITE,
										  LH_OVERFLOW_PAGE,
										  fc_bstrategy);
		fc_rpage = BufferGetPage(fc_rbuf);
		fc_ropaque = HashPageGetOpaque(fc_rpage);
		Assert(fc_ropaque->hasho_bucket == fc_bucket);
	}

	/* 不可达 */
}
